package com.shujia.flink.core

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

object Demo3EventTime {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    /**
      * 1、设置时间模式为事件时间
      *
      */
    //默认是处理时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)


    env.setParallelism(1)


    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)


    /**
      * 每隔5秒统计单词的数量
      *
      */

    val wordsDS: DataStream[(String, Long)] = linesDS.map(line => {
      val split: Array[String] = line.split(",")
      (split(0), split(1).toLong)
    })

    /*


java,1642494546000
java,1642494547000
java,1642494548000
java,1642494550000
java,1642494549000
java,1642494551000
java,1642494552000
java,1642494553000
java,1642494554000
java,1642494555000
java,1642494560000

      *
      */
    /**
      * 2、指定时间字段, 必须是时间戳毫秒级别
      *
      */

    //水位线默认等于时间戳最大的数据的时间
    //val assDS: DataStream[(String, Long)] = wordsDS.assignAscendingTimestamps(_._2)

    /**
      * 指定时间字段和水位线
      *
      */
    val assDS: DataStream[(String, Long)] = wordsDS.assignTimestampsAndWatermarks(
      //指定时间戳字段和数据最大乱序时间
      new BoundedOutOfOrdernessTimestampExtractor[(String, Long)](Time.seconds(5)) {
        override def extractTimestamp(element: (String, Long)): Long = element._2
      }
    )

    val countDS: DataStream[(String, Int)] =
      assDS
        .map(kv => (kv._1, 1))
        .keyBy(_._1)
        .timeWindow(Time.seconds(5))
        .sum(1)

    countDS.print()


    env.execute()

  }

}
